package com.atguigu.gmall.realtime.dws.app;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.common.base.BaseApp;
import com.atguigu.gmall.realtime.common.bean.TradeSkuOrderBean;
import com.atguigu.gmall.realtime.common.constant.Constant;
import com.atguigu.gmall.realtime.common.function.BeanToJsonStrMapFunction;
import com.atguigu.gmall.realtime.common.function.DimAsyncFunction;
import com.atguigu.gmall.realtime.common.util.DateFormatUtil;
import com.atguigu.gmall.realtime.common.util.FlinkSinkUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.math.BigDecimal;
import java.util.concurrent.TimeUnit;

/**
 * @author Felix
 * @date 2024/10/13
 * SKU粒度下单聚合统计
 * 需要启动的进程
 *      zk、kafka、maxwell、hdfs、hbase、redis、doris、DwdTradeOrderDetail、DwsTradeSkuOrderWindow
 * 开发流程
 *      从kafka的下单事实表中读取数据
 *      类型转换并过滤空消息  jsonStr->jsonObj
 *      去重
 *          为什么会产生空消息以及重复数据？
 *              dwd下单事实表是由4张表组成，订单明细表、订单表、订单明细活动表、订单明细优惠券表
 *              这4张表进行关联的时候，订单明细是主表，和订单使用的是内连接
 *              和订单明细活动以及订单明细优惠券使用的左外连接
 *              如果左外连接，左表数据先到，右表数据后到，会产生3条数据
 *                  左表      null    +I
 *                  左表      null    -D
 *                  左表      右表    +I
 *              这样的数据如果发送到kafka主题，kafka主题会接收到3条消息
 *                  左表  null
 *                  null
 *                  左表  右表
 *              如果使用FlinkSQL方式从kafka主题中读取数据，会自定的对空消息进行过滤
 *              如果使用FlinikAPI方式从kafka主题中读取数据，默认的SimpleStringSchema是处理不了空消息的，在创建KafkaSource对象的时候，需要自定义反序列化器
 *              在程序中需要将空消息过滤掉并对第1条数据和第3条数据进行去重
 *
 *          按照唯一键（订单明细id）进行分组
 *          去重方式1：状态 +定时器
 *              优点：如果出现重复了，也只会向下游发送一条数据  缺点：不管是否重复，都需要等5s才会向下游传递数据，时效性差
 *              当第1条数据到来的时候，将数据放到状态中，并注册5s后执行的定时器
 *              当第2条重复数据到来的时候，会用第2条数据的聚合时间和状态中存放的第1条数据的聚合时间进行比较，将时间大的放到状态中
 *              当定时器被触发执行的时候，将状态中的数据发送到下游
 *          去重方式2：状态 + 抵消
 *              优点：时效性好     缺点：如果出现重复了，会向下游传递3条数据（数据膨胀）
 *              当第1条数据到来的时候，将数据放到状态中，并直接向下游传递一次
 *              当第2条重复数据到来的时候，会将状态中影响到度量值的字段进行取反，再向下游传递
 *              将第2条数据也向下游传递
 *
 *      再次对流中数据进行转换 jsonObj->统计的实体类对象
 *      指定Watermark以及提取事件时间字段
 *      按照统计的为sku进行分组
 *      开窗
 *      聚合
 *      关联维度
 *          V1.0
 *              维度关联最基本实现
 *              HBaseUtil-->getRow()
 *          V2.0
 *              旁路缓存
 *                  思路：先从缓存中获取维度数据，如果从缓存中找到了要关联的维度，直接将其作为返回值返回(缓存命中)
 *                      如果在缓存中没有找到要关联的维度，发送请求到HBase中进行查找
 *                      并将查到的维度数据放到缓存中缓存起来，方便下次查询使用
 *                  选型
 *                      状态      性能很好    维护性差
 *                      Redis    性能不错    维护性好   √
 *                  关于Redis的一些配置
 *                      key :   维度表名:主键值
 *                      type:   String
 *                      expire: 1day    避免冷数据常驻内存，给内存带来压力
 *                      注意：   如果维度数据发生变化，将缓存数据清除掉       DimSinkFunction->invoke->jedis.del()
 *                  RedisUtil->readDim、writeDim
 *          V3.0
 *              抽取旁路缓存模板
 *              class DimMapFunction
 *          V4.0
 *              旁路缓存 + 异步IO
 *              为什么使用异步IO？
 *                  在使用外部系统数据，扩充流中数据的时候(维度关联)，如果要向提升处理能力，可以调大并行度，但是更大的并行度意味需要更多的硬件资源，
 *                  不可能无限制的调整。
 *                  在硬件资源有限的情况下，可以使用异步的方式对流中单个并行度的数据进行处理。
 *              Flink提供的专门发送异步请求的API
 *                  AsyncDataStream.[un]OrderedWait(
 *                      流,
 *                      如何发送异步请求, 需要实现AsyncFunction,
 *                      超时时间,
 *                      时间单位
 *                  )
 *              要向实现异步的维度关联
 *                  RedisUtil ->提供了异步操作Redis的方法
 *                  HBaseUtil ->提供了异步操作HBase的方法
 *
 *          V5.0
 *              旁路缓存 + 异步IO + 模板
 *              class DimAsyncFunction{
 *                  asynvInvoke{
 *                      //创建异步编排对象，有返回值
 *                          从Redis中读取数据
 *                      //创建新的线程任务，有入参，有返回值   上一个线程任务的返回值会作为当前线程任务的入参
 *                          从HBase中读取数据，并往Redis放
 *                      //创建新的线程任务，有入参，无返回值
 *                          维度属性的补充
 *                  }
 *              }
 *
 *      将关联的结果写到Doris
 */
public class DwsTradeSkuOrderWindow extends BaseApp {
    public static void main(String[] args) {
        new DwsTradeSkuOrderWindow().start(
                10029,
                4,
                "dws_trade_sku_order_window",
                Constant.TOPIC_DWD_TRADE_ORDER_DETAIL
        );

    }
    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> kafkaStrDS) {
        //TODO 1.对流中数据进行类型转换并过滤空消息    jsonStr->jsonObj
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaStrDS.process(
                new ProcessFunction<String, JSONObject>() {
                    @Override
                    public void processElement(String jsonStr, ProcessFunction<String, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {
                        if (StringUtils.isNotEmpty(jsonStr)) {
                            JSONObject jsonObj = JSON.parseObject(jsonStr);
                            out.collect(jsonObj);
                        }
                    }
                }
        );
        //jsonObjDS.print();
        //TODO 2.去重
        //2.1 按照唯一键(订单明细id)进行分组
        KeyedStream<JSONObject, String> orderDetailIdKeyedDS
                = jsonObjDS.keyBy(jsonObj -> jsonObj.getString("id"));
        //2.2 去重方式1：状态 + 定时器      优点：如果出现重复，只会向下游发送一条数据，不会膨胀  缺点:不管数据是否重复，都要等5s后才能向下游传递，影响时效性
        /*
        SingleOutputStreamOperator<JSONObject> distinctDS = orderDetailIdKeyedDS.process(
                new KeyedProcessFunction<String, JSONObject, JSONObject>() {
                    ValueState<JSONObject> lastJsonObjState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        ValueStateDescriptor<JSONObject> valueStateDescriptor
                                = new ValueStateDescriptor<JSONObject>("lastJsonObjState", JSONObject.class);
                        lastJsonObjState = getRuntimeContext().getState(valueStateDescriptor);
                    }

                    @Override
                    public void processElement(JSONObject jsonObj, KeyedProcessFunction<String, JSONObject, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {
                        //从状态中获取上条数据
                        JSONObject lastJsonObj = lastJsonObjState.value();
                        if (lastJsonObj == null) {
                            //说明没有出现重复
                            //将当前数据放到状态中
                            lastJsonObjState.update(jsonObj);
                            //注册一个5s后执行的定时器
                            TimerService timerService = ctx.timerService();
                            long currentProcessingTime = timerService.currentProcessingTime();
                            timerService.registerProcessingTimeTimer(currentProcessingTime + 5000L);
                        } else {
                            //说明重复了
                            //获取先来的数据的关联时间  伪代码
                            String ts1 = lastJsonObj.getString("关联时间属性");
                            //获取当前的数据的关联时间
                            String ts2 = jsonObj.getString("关联时间属性");
                            if (ts2.compareTo(ts1) >= 0) {
                                //将关联时间大的数据放到状态中
                                lastJsonObjState.update(jsonObj);
                            }
                        }
                    }

                    @Override
                    public void onTimer(long timestamp, KeyedProcessFunction<String, JSONObject, JSONObject>.OnTimerContext ctx, Collector<JSONObject> out) throws Exception {
                        //定时器被触发的时候执行的方法
                        //从状态中获取数据
                        JSONObject jsonObj = lastJsonObjState.value();
                        //将数据发送到下游
                        out.collect(jsonObj);
                        //清状态
                        lastJsonObjState.clear();
                    }
                }
        );
        */
        //2.3 去重方式2：状态 + 抵消     优点：时效性好     缺点：如果出现重复，需要向下游发送3条数据(数据膨胀)
        SingleOutputStreamOperator<JSONObject> distinctDS = orderDetailIdKeyedDS.process(
                new KeyedProcessFunction<String, JSONObject, JSONObject>() {
                    ValueState<JSONObject> lastJsonObjState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        ValueStateDescriptor<JSONObject> valueStateDescriptor
                                = new ValueStateDescriptor<JSONObject>("lastJsonObjState", JSONObject.class);
                        valueStateDescriptor.enableTimeToLive(StateTtlConfig.newBuilder(Time.seconds(5)).build());
                        lastJsonObjState = getRuntimeContext().getState(valueStateDescriptor);
                    }

                    @Override
                    public void processElement(JSONObject jsonObj, KeyedProcessFunction<String, JSONObject, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {
                        //从状态中获取上条数据
                        JSONObject lastJsonObj = lastJsonObjState.value();
                        if (lastJsonObj != null) {
                            //说明重复了  将状态中的这条数据影响到度量值的字段进行取反再向下游发送一次
                            String splitOriginalAmount = lastJsonObj.getString("split_original_amount");
                            String splitCouponAmount = lastJsonObj.getString("split_coupon_amount");
                            String splitActivityAmount = lastJsonObj.getString("split_activity_amount");
                            String splitTotalAmount = lastJsonObj.getString("split_total_amount");

                            lastJsonObj.put("split_original_amount", "-" + splitOriginalAmount);
                            lastJsonObj.put("split_coupon_amount", "-" + splitCouponAmount);
                            lastJsonObj.put("split_activity_amount", "-" + splitActivityAmount);
                            lastJsonObj.put("split_total_amount", "-" + splitTotalAmount);

                            out.collect(lastJsonObj);
                        }
                        lastJsonObjState.update(jsonObj);
                        out.collect(jsonObj);
                    }
                }
        );
        //distinctDS.print();

        //TODO 3.再次对流中数据进行类型转换  jsonObj->统计的实体类对象  相当于WordCount将单词转换为二元组
        //{"create_time":"2024-10-12 15:21:11","sku_num":"1","activity_rule_id":"1","split_original_amount":"6499.0000","split_coupon_amount":"0.0",
        // "sku_id":"3","date_id":"2024-10-12","user_id":"2881","province_id":"23","activity_id":"1","sku_name":"小米12S ","id":"15106",
        // "order_id":"10668","split_activity_amount":"500.0","split_total_amount":"5999.0","ts":1728804071}
        SingleOutputStreamOperator<TradeSkuOrderBean> beanDS = distinctDS.map(
                new MapFunction<JSONObject, TradeSkuOrderBean>() {
                    @Override
                    public TradeSkuOrderBean map(JSONObject jsonObj) throws Exception {
                        String skuId = jsonObj.getString("sku_id");
                        BigDecimal splitOriginalAmount = jsonObj.getBigDecimal("split_original_amount");
                        BigDecimal splitCouponAmount = jsonObj.getBigDecimal("split_coupon_amount");
                        BigDecimal splitActivityAmount = jsonObj.getBigDecimal("split_activity_amount");
                        BigDecimal splitTotalAmount = jsonObj.getBigDecimal("split_total_amount");
                        Long ts = jsonObj.getLong("ts") * 1000;

                        TradeSkuOrderBean orderBean = TradeSkuOrderBean.builder()
                                .skuId(skuId)
                                .originalAmount(splitOriginalAmount)
                                .couponReduceAmount(splitCouponAmount)
                                .activityReduceAmount(splitActivityAmount)
                                .orderAmount(splitTotalAmount)
                                .ts(ts)
                                .build();
                        return orderBean;
                    }
                }
        );
        //beanDS.print();

        //TODO 4.指定Watermark的生成策略以及提取事件时间字段
        SingleOutputStreamOperator<TradeSkuOrderBean> withWatermarkDS = beanDS.assignTimestampsAndWatermarks(
                WatermarkStrategy
                        .<TradeSkuOrderBean>forMonotonousTimestamps()
                        .withTimestampAssigner(
                                new SerializableTimestampAssigner<TradeSkuOrderBean>() {
                                    @Override
                                    public long extractTimestamp(TradeSkuOrderBean bean, long recordTimestamp) {
                                        return bean.getTs();
                                    }
                                }
                        )
        );
        //TODO 5.按照统计的维度sku进行分组
        KeyedStream<TradeSkuOrderBean, String> skuIdKeyedDS = withWatermarkDS.keyBy(TradeSkuOrderBean::getSkuId);

        //TODO 6.开窗
        WindowedStream<TradeSkuOrderBean, String, TimeWindow> windowDS
                = skuIdKeyedDS.window(TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(10)));
        //TODO 7.聚合
        SingleOutputStreamOperator<TradeSkuOrderBean> reduceDS = windowDS.reduce(
                new ReduceFunction<TradeSkuOrderBean>() {
                    @Override
                    public TradeSkuOrderBean reduce(TradeSkuOrderBean value1, TradeSkuOrderBean value2) throws Exception {
                        value1.setOriginalAmount(value1.getOriginalAmount().add(value2.getOriginalAmount()));
                        value1.setCouponReduceAmount(value1.getCouponReduceAmount().add(value2.getCouponReduceAmount()));
                        value1.setActivityReduceAmount(value1.getActivityReduceAmount().add(value2.getActivityReduceAmount()));
                        value1.setOrderAmount(value1.getOrderAmount().add(value2.getOrderAmount()));
                        return value1;
                    }
                },
                new WindowFunction<TradeSkuOrderBean, TradeSkuOrderBean, String, TimeWindow>() {
                    @Override
                    public void apply(String s, TimeWindow window, Iterable<TradeSkuOrderBean> input, Collector<TradeSkuOrderBean> out) throws Exception {
                        TradeSkuOrderBean orderBean = input.iterator().next();
                        orderBean.setStt(DateFormatUtil.tsToDateTime(window.getStart()));
                        orderBean.setEdt(DateFormatUtil.tsToDateTime(window.getEnd()));
                        orderBean.setCurDate(DateFormatUtil.tsToDate(window.getStart()));
                        out.collect(orderBean);
                    }
                }
        );
        //reduceDS.print();
        //TODO 8.关联SKU维度
        /*
        //V1.0  维度关联的最基本的实现方式
        SingleOutputStreamOperator<TradeSkuOrderBean> withSkuInfoDS = reduceDS.map(
                new RichMapFunction<TradeSkuOrderBean, TradeSkuOrderBean>() {

                    Connection hbaseConn;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        hbaseConn = HBaseUtil.getHBaseConnection();
                    }

                    @Override
                    public void close() throws Exception {
                        HBaseUtil.closeHBaseConnection(hbaseConn);
                    }

                    @Override
                    public TradeSkuOrderBean map(TradeSkuOrderBean orderBean) throws Exception {
                        //获取要关联的维度主键
                        String skuId = orderBean.getSkuId();
                        //根据维度主键到HBase表中获取对应的维度数据
                        //{"is_sale":"1","sku_desc":"小米12S","create_time":"2021-12-14 00:00:00","tm_id":"5","price":"6999",
                        // "sku_default_img":"http://47.93.148.192:8080/group1/M00/00/01/rBHu8l-rfvmAIpgZAAIvrX6L9fo612.jpg",
                        // "weight":"1.00","sku_name":"小米12S","id":"1","spu_id":"1","category3_id":"61"}
                        JSONObject dimJsonObj = HBaseUtil.getRow(hbaseConn, Constant.HBASE_NAMESPACE, "dim_sku_info", skuId, JSONObject.class);

                        //将维度属性补充到流中对象上
                        orderBean.setSkuName(dimJsonObj.getString("sku_name"));
                        orderBean.setSpuId(dimJsonObj.getString("spu_id"));
                        orderBean.setCategory3Id(dimJsonObj.getString("category3_id"));
                        orderBean.setTrademarkId(dimJsonObj.getString("tm_id"));
                        return orderBean;
                    }
                }
        );
        withSkuInfoDS.print();

        //V2.0 旁路缓存优化
        SingleOutputStreamOperator<TradeSkuOrderBean> withSkuInfoDS = reduceDS.map(
                new RichMapFunction<TradeSkuOrderBean, TradeSkuOrderBean>() {
                    private Connection hbaseConn;
                    private Jedis jedis;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        hbaseConn = HBaseUtil.getHBaseConnection();
                        jedis = RedisUtil.getJedis();
                    }

                    @Override
                    public void close() throws Exception {
                        HBaseUtil.closeHBaseConnection(hbaseConn);
                        RedisUtil.closeJedis(jedis);
                    }

                    @Override
                    public TradeSkuOrderBean map(TradeSkuOrderBean orderBean) throws Exception {
                        //根据流中对象获取要关联的维度的主键
                        String skuId = orderBean.getSkuId();
                        //根据维度主键到Redis中获取要关联的维度数据
                        JSONObject dimJsonObj = RedisUtil.readDim(jedis, "dim_sku_info", skuId);
                        if (dimJsonObj != null) {
                            //如果从Redis获取到了要关联的维度，直接将其作为返回值返回(缓存命中)
                            System.out.println("~~~从Redis中获取维度数据~~~");
                        } else {
                            //如果从Redis中没有获取到要关联的维度，发送请求到HBase中查询维度
                            dimJsonObj = HBaseUtil.getRow(hbaseConn, Constant.HBASE_NAMESPACE, "dim_sku_info", skuId, JSONObject.class);
                            if (dimJsonObj != null) {
                                System.out.println("~~~从HBase中获取维度数据~~~");
                                //将查询结果放到Redis中缓存起来，方便下次查询使用
                                RedisUtil.writeDim(jedis, "dim_sku_info", skuId, dimJsonObj);
                            } else {
                                System.out.println("~~~没有找到要关联的维度~~~");
                            }
                        }
                        //将维度属性补充到流中对象上
                        if(dimJsonObj != null){
                            orderBean.setSkuName(dimJsonObj.getString("sku_name"));
                            orderBean.setSpuId(dimJsonObj.getString("spu_id"));
                            orderBean.setCategory3Id(dimJsonObj.getString("category3_id"));
                            orderBean.setTrademarkId(dimJsonObj.getString("tm_id"));
                        }
                        return orderBean;
                    }
                }
        );

        //withSkuInfoDS.print();

        //V3.0 旁路缓存 + 模板
        SingleOutputStreamOperator<TradeSkuOrderBean> withSkuInfoDS = reduceDS.map(
                new DimMapFunction<TradeSkuOrderBean>() {
                    @Override
                    public void addDims(TradeSkuOrderBean orderBean, JSONObject dimJsonObj) {
                        orderBean.setSkuName(dimJsonObj.getString("sku_name"));
                        orderBean.setSpuId(dimJsonObj.getString("spu_id"));
                        orderBean.setCategory3Id(dimJsonObj.getString("category3_id"));
                        orderBean.setTrademarkId(dimJsonObj.getString("tm_id"));
                    }

                    @Override
                    public String getTableName() {
                        return "dim_sku_info";
                    }

                    @Override
                    public String getRowKey(TradeSkuOrderBean orderBean) {
                        return orderBean.getSkuId();
                    }
                }
        );

        withSkuInfoDS.print();

        //V4.0 旁路缓存 + 异步IO
        //将异步I/O操作应用于DataStream作为DataStream的一次转换操作
        SingleOutputStreamOperator<TradeSkuOrderBean> withSkuInfoDS = AsyncDataStream.unorderedWait(
                reduceDS,
                //如何发送异步请求，需要实现AsyncFunction
                new RichAsyncFunction<TradeSkuOrderBean, TradeSkuOrderBean>() {
                    StatefulRedisConnection<String,String> asyncRedisConn;
                    AsyncConnection asyncHBaseConn;
                    @Override
                    public void open(Configuration parameters) throws Exception {
                        asyncRedisConn = RedisUtil.getRedisAsyncConnection();
                        asyncHBaseConn = HBaseUtil.getHBaseAsyncConnection();
                    }

                    @Override
                    public void close() throws Exception {
                        RedisUtil.closeRedisAsyncConnection(asyncRedisConn);
                        HBaseUtil.closeHBaseAsyncConnection(asyncHBaseConn);
                    }

                    @Override
                    public void asyncInvoke(TradeSkuOrderBean orderBean, ResultFuture<TradeSkuOrderBean> resultFuture) throws Exception {
                        //根据流中对象获取要关联的维度的主键
                        String skuId = orderBean.getSkuId();
                        //以异步的方式从Redis中获取维度数据
                        JSONObject dimJsonObj = RedisUtil.readDimAsync(asyncRedisConn, "dim_sku_info", skuId);
                        if(dimJsonObj != null){
                            //如果在Redis中获取到了对应的维度，直接将其作为返回值进行返回(缓存命中)
                            System.out.println("~~~从Redis中获取到了维度数据~~~");
                        }else{
                            //如果在Redis中没有获取到对应的维度，以异步的方式到HBase中查询
                            dimJsonObj = HBaseUtil.readDimAsync(asyncHBaseConn,Constant.HBASE_NAMESPACE,"dim_sku_info", skuId);
                            if(dimJsonObj != null){
                                System.out.println("~~~从HBase中获取到了维度数据~~~");
                                //以异步的方式将查询到维度数据放到Redis中进行缓存，方便下次查询使用
                                RedisUtil.writeDimAsync(asyncRedisConn,"dim_sku_info", skuId,dimJsonObj);
                            }else{
                                System.out.println("~~~没有找到要关联的维度~~~");
                            }
                        }
                        //将维度属性补充到流中对象上
                        if(dimJsonObj != null){
                            orderBean.setSkuName(dimJsonObj.getString("sku_name"));
                            orderBean.setSpuId(dimJsonObj.getString("spu_id"));
                            orderBean.setCategory3Id(dimJsonObj.getString("category3_id"));
                            orderBean.setTrademarkId(dimJsonObj.getString("tm_id"));
                        }
                        //获取数据库交互的结果并发送给 ResultFuture 的 回调 函数
                        resultFuture.complete(Collections.singleton(orderBean));
                    }
                },
                60,
                TimeUnit.SECONDS
        );
        //withSkuInfoDS.print();
         */
        //V5.0 旁路缓存 + 异步IO + 模板方法
        SingleOutputStreamOperator<TradeSkuOrderBean> withSkuInfoDS = AsyncDataStream.unorderedWait(
                reduceDS,
                new DimAsyncFunction<TradeSkuOrderBean>() {
                    @Override
                    public void addDims(TradeSkuOrderBean orderBean, JSONObject dimJsonObj) {
                        orderBean.setSkuName(dimJsonObj.getString("sku_name"));
                        orderBean.setSpuId(dimJsonObj.getString("spu_id"));
                        orderBean.setCategory3Id(dimJsonObj.getString("category3_id"));
                        orderBean.setTrademarkId(dimJsonObj.getString("tm_id"));
                    }

                    @Override
                    public String getTableName() {
                        return "dim_sku_info";
                    }

                    @Override
                    public String getRowKey(TradeSkuOrderBean orderBean) {
                        return orderBean.getSkuId();
                    }
                },
                60,
                TimeUnit.SECONDS
        );
        //withSkuInfoDS.print();

        //TODO 9.关联SPU维度
        SingleOutputStreamOperator<TradeSkuOrderBean> withSpuInfoDS = AsyncDataStream.unorderedWait(
                withSkuInfoDS,
                new DimAsyncFunction<TradeSkuOrderBean>() {
                    @Override
                    public void addDims(TradeSkuOrderBean orderBean, JSONObject dimJsonObj) {
                        orderBean.setSpuName(dimJsonObj.getString("spu_name"));
                    }

                    @Override
                    public String getTableName() {
                        return "dim_spu_info";
                    }

                    @Override
                    public String getRowKey(TradeSkuOrderBean orderBean) {
                        return orderBean.getSpuId();
                    }
                },
                60,
                TimeUnit.SECONDS
        );

        //TODO 10.关联TM维度
        SingleOutputStreamOperator<TradeSkuOrderBean> withTmDS = AsyncDataStream.unorderedWait(
                withSpuInfoDS,
                new DimAsyncFunction<TradeSkuOrderBean>() {
                    @Override
                    public void addDims(TradeSkuOrderBean orderBean, JSONObject dimJsonObj) {
                        orderBean.setTrademarkName(dimJsonObj.getString("tm_name"));
                    }

                    @Override
                    public String getTableName() {
                        return "dim_base_trademark";
                    }

                    @Override
                    public String getRowKey(TradeSkuOrderBean orderBean) {
                        return orderBean.getTrademarkId();
                    }
                },
                60,
                TimeUnit.SECONDS
        );
        //TODO 11.关联Category3维度
        SingleOutputStreamOperator<TradeSkuOrderBean> c3Stream = AsyncDataStream.unorderedWait(
                withTmDS,
                new DimAsyncFunction<TradeSkuOrderBean>() {
                    @Override
                    public String getRowKey(TradeSkuOrderBean bean) {
                        return bean.getCategory3Id();
                    }

                    @Override
                    public String getTableName() {
                        return "dim_base_category3";
                    }

                    @Override
                    public void addDims(TradeSkuOrderBean bean, JSONObject dim) {
                        bean.setCategory3Name(dim.getString("name"));
                        bean.setCategory2Id(dim.getString("category2_id"));
                    }
                },
                120,
                TimeUnit.SECONDS
        );

        //TODO 12.关联Category2维度
        SingleOutputStreamOperator<TradeSkuOrderBean> c2Stream = AsyncDataStream.unorderedWait(
                c3Stream,
                new DimAsyncFunction<TradeSkuOrderBean>() {
                    @Override
                    public String getRowKey(TradeSkuOrderBean bean) {
                        return bean.getCategory2Id();
                    }

                    @Override
                    public String getTableName() {
                        return "dim_base_category2";
                    }

                    @Override
                    public void addDims(TradeSkuOrderBean bean, JSONObject dim) {
                        bean.setCategory2Name(dim.getString("name"));
                        bean.setCategory1Id(dim.getString("category1_id"));
                    }
                },
                120,
                TimeUnit.SECONDS
        );

        //TODO 13.关联Category1维度
        SingleOutputStreamOperator<TradeSkuOrderBean> withC1DS = AsyncDataStream.unorderedWait(
                c2Stream,
                new DimAsyncFunction<TradeSkuOrderBean>() {
                    @Override
                    public String getRowKey(TradeSkuOrderBean bean) {
                        return bean.getCategory1Id();
                    }

                    @Override
                    public String getTableName() {
                        return "dim_base_category1";
                    }

                    @Override
                    public void addDims(TradeSkuOrderBean bean, JSONObject dim) {
                        bean.setCategory1Name(dim.getString("name"));
                    }
                },
                120,
                TimeUnit.SECONDS
        );

        //TODO 14.将关联的结果写到Doris
        withC1DS.print();
        withC1DS
                .map(new BeanToJsonStrMapFunction<>())
                .sinkTo(FlinkSinkUtil.getDorisSink("dws_trade_sku_order_window"));


    }
}
